//
// Created by Administrator on 2024/9/26.
//

#ifndef SERVER_HPP
#define SERVER_HPP

#include <iostream>
#include <boost/asio.hpp>
#include <utility>

#include "common.hpp"
#include "packet_handler.hpp"
#include "waitable.hpp"

namespace asio = boost::asio;
using asio::ip::tcp;
using boost::system::error_code;

namespace network {
namespace details {


const std::set<uint16_t> allowed_messages_before_login{0x0100, 0x0102};
constexpr uint16_t verify_message_id = 0x0102;
constexpr uint16_t logout_message_id = 0x0003;
/**
 * 长连接会话
 */
class Session final : public std::enable_shared_from_this<Session>, public Waitable<protocol::platform::CommonResponse> {
public:
    explicit Session(asio::io_context &io_context,
        notifier_type<std::shared_ptr<Session>, const std::string&> notifier)
        : io_context_(io_context),socket_(io_context), disconnect_timer_(io_context),notifier_(std::move(notifier)) { }

    tcp::socket& socket() {
        return socket_;
    }

    /**
     * 建立socket连接会话
     */
    void start() {
        std::stringstream peer;
        peer << socket_.remote_endpoint();
        endpoint_ = peer.str();
        INFO(logger, "Connection with 【{}】 has been built", endpoint_);
        packet_handler_ = PacketHandler::create(endpoint_, [&](const protocol::Packet &packet) {receive_handler(packet);});
        reset_disconnect_timer();
        start_read();
    }

    /**
     * 发送消息到远端，会进行消息重传
     * @param to_reply 要发送的消息
     **/
    void send(const protocol::Packet &to_reply) {
        if (!to_reply) return;
        deliver(to_reply);
        std::thread(&Session::retransfer_handler, this, to_reply).detach();
    }

    ~Session() override = default;

    // 关闭连接
    void close_connection() {
        INFO(logger, "Connection with 【{}】 will be closed", endpoint_);
        try {
            disconnect_timer_.cancel();
            verified = false;
            socket_.shutdown(asio::socket_base::shutdown_both);
            socket_.close();
            // 通知会话管理者已断开连接
            notifier_(notify_type::disconnect, shared_from_this(), connected_terminal);
        } catch (std::exception &e) {
            ERR(logger, e.what());
        }
    }

    // 远端地址 (ip:port)
    std::string connected_terminal;

private:
    /**
     * 向客户端发送消息
     * @param packet 要发送的消息
     */
    void deliver(const protocol::Packet &packet) {
        if (!packet) return;
        auto token = std::bind(&Session::send_handler,
                    shared_from_this(), asio::placeholders::error, asio::placeholders::bytes_transferred);
        try {
            const auto sld = packet_handler_->segment(*packet);
            for (auto to_send : sld) {
                socket_.async_send(asio::const_buffer(to_send->data, to_send->size), token);
            }
        } catch (std::invalid_argument&) {
            const auto sld = packet->serialize();
            socket_.async_send(asio::const_buffer(sld->data, sld->size), token);
        } catch (std::exception& e) {
            ERR(logger, e.what());
        }
    }

    /**
     * 等待指定消息是否收到回复，未收到进行重传，直到最大重传次数
     * @param packet 要检查和重传的包
     * TODO 最大重传次数和初始等待时间从终端配置中获取
     */
    void retransfer_handler(const protocol::Packet packet) {
        uint16_t wait;
        bool wait_sn = false;
        if (wait_sn = protocol::MessageFactory::messages_with_sn_response.contains(packet->header_.message_id_)) {
            wait = packet->header_.message_sn_;
        } else if (protocol::MessageFactory::response_message_id_map.contains(packet->header_.message_id_)) {
            wait = protocol::MessageFactory::response_message_id_map.at(packet->header_.message_id_);
        } else {
            return;
        }

        std::chrono::seconds wait_time(BASIC_RETRANSFER_TIME);
        int count = 0;
        for (count = 0; count < MAX_RETRANSFER_COUNT; ++count) {
            wait_time = wait_time*(count+1);
            if (wait_sn ? wait_for(wait, wait_time).first : wait_for_next(wait, wait_time).first)
                break;
            deliver(packet);
        }
        if (count >= MAX_RETRANSFER_COUNT) {
            ERR(logger, "A meesage whose sn={} and id={} has not been responsed by remote=【{}】",
                          (int) packet->header_.message_sn_, common::toHex(&packet->header_.message_id_, 2), endpoint_);
        }
    }

    /**
     * 对接收到的包进行处理和回复
     * @param packet 接收到的包
     */
    void receive_handler(protocol::Packet packet) {
        auto &hdr = packet->header_;
        if (!verified && !allowed_messages_before_login.contains(hdr.message_id_)) {
            logger->debug("An invalid message whose id={} has been rejected", hdr.message_id_);
            return;
        }
        if (protocol::MessageFactory::response_with_source_sn.contains(hdr.message_id_))
            get_it(packet, false);
        get_it(packet, true);
        INFO(logger, "读取到消息：\n{}", packet->toString());
        reset_disconnect_timer();

        Gateway gateway(packet, packet_handler_);
        if (const auto to_send = gateway.route()) {
            if (hdr.message_id_ == verify_message_id) {
                auto tosend_body = dynamic_cast<protocol::platform::CommonResponse *>(to_send->body_.get());
                verified = tosend_body->result_ == (uint16_t)protocol::platform::ResponseResult::SUCCESS;
                if (verified) {
                    connected_terminal = common::toHex(packet_handler_->terminal_imei.data(),
                                            packet_handler_->terminal_imei.size());
                    notifier_(notify_type::verified, shared_from_this(), connected_terminal);
                }
            }
            else if (hdr.message_id_ == logout_message_id)
                verified = dynamic_cast<
                           protocol::platform::CommonResponse *>(to_send->body_.get())->result_ !=
                       (uint16_t)protocol::platform::ResponseResult::SUCCESS;
            send(to_send);
        }
    }

    void send_handler(const error_code& ec, const size_t &bytes_transferred) const {
        if (ec) {
            ERR(logger, "Failed to send message because {}", ec.message());
        }
    }

    /**
     * 重置未保活断连计时器
     */
    void reset_disconnect_timer() {
        disconnect_timer_.expires_after(TIME_TO_DISCONNECT);
        disconnect_timer_.async_wait(std::bind(&Session::handle_time_out, shared_from_this(),
            asio::placeholders::error));
    }

    void start_read() {
        socket_.async_read_some(asio::buffer(*read_buffer_, 102400),
            std::bind(&Session::handle_read, shared_from_this(),
                asio::placeholders::error, asio::placeholders::bytes_transferred));
    }

    /**
     * 收到消息时回调
     * @param ec 错误码
     * @param bytes_transferred 读到了多少字节
     */
    void handle_read(const error_code& ec, const size_t &bytes_transferred) {
        if (!ec) {
            logger->debug("read {} bytes", bytes_transferred);
            logger->debug("read: {}", common::toHex(read_buffer_->data(), bytes_transferred));
            packet_handler_->async_handle(std::make_shared<common::Serialized>(read_buffer_->data(), read_buffer_->size()));
            start_read();
        } else if (ec == asio::error::connection_reset) {
            WARN(logger, "client from 【{}】 close connection forcely", endpoint_);
            close_connection();
        } else if (ec == asio::error::eof) {
            WARN(logger, "client from 【{}】 close connection", endpoint_);
            close_connection();
        } else if (ec == asio::error::shut_down) {
            WARN(logger, "socket has been shutdown");
        } else if (ec == asio::error::operation_aborted) {
            WARN(logger, "operation aborted");
        } else {
            ERR(logger, "other error:{}",ec.what());
            close_connection();
        }

    }

    // 长期未收到消息时断开连接
    void handle_time_out(const error_code &ec) {
        if (!ec)
            close_connection();
        else if (ec == asio::error::operation_aborted) {
            if (verified)
                logger->debug("disconnect timer has been reset");
            else
                logger->debug("disconnect timer has been canceled");
        }
    }

    asio::io_context &io_context_;
    tcp::socket socket_;
    asio::steady_timer disconnect_timer_;
    notifier_type<std::shared_ptr<Session>, const std::string&> notifier_;

    std::shared_ptr<std::vector<char>> read_buffer_ = std::make_shared<std::vector<char>>(1024);
    std::string endpoint_;
    std::string unknown_ip;
    std::string host_, port_;
    bool verified = false;

    handler_t packet_handler_;

    logger_t logger = common::Logger::logger("Session");
};
}

using details::Session;
class Server {
public:
    Server(asio::io_context &io_context, const uint16_t port)
        : io_context_(io_context), acceptor_(io_context, tcp::endpoint(tcp::v4(), port)) {
        logger->info("The server is listening on port {}", std::to_string(port));
        start_accept();
    }

private:
    bool notify_func(notify_type nt, std::shared_ptr<Session> who_notify_me, const std::string &param) {
        if (notify_type::disconnect == nt) {
            if (connected_clients.erase(param) == 0) { // 要断开的连接不在已连接记录中
                logger->info("There is no login info");
                return false;
            }
            return true;
        } else if (notify_type::verified == nt) {
            if (connected_clients.contains(param) && connected_clients[param]->socket().remote_endpoint() != who_notify_me->socket().remote_endpoint()) {
                WARN(logger, "client whose imei = {} already verified, but a new one try to verify again", param);
                connected_clients[param]->close_connection();
            }
            connected_clients[param] = who_notify_me;
        }
        return true;
    }

    void start_accept() {
        auto session = std::make_shared<Session>(std::ref(io_context_),
                                                 std::bind(&Server::notify_func, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        acceptor_.async_accept(session->socket(),
            std::bind(&Server::handle_accept, this, session,asio::placeholders::error));
    }

    void handle_accept(std::shared_ptr<Session> new_session,
        const error_code &ec) {
        if (!ec) {
            new_session->start();
        } else {
            logger->error("An error found: {}", ec.what());
        }
        start_accept();
    }

    asio::io_context &io_context_;
    tcp::acceptor acceptor_;
    std::unordered_map<std::string, std::shared_ptr<Session>> connected_clients;
    std::function<void(std::shared_ptr<common::Container<char>>, std::shared_ptr<Session>)> sender_;

    logger_t logger = common::Logger::logger("Server");
};

}


#endif //SERVER_HPP
